package tech.yunqian.slim.project.controller;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import com.alibaba.fastjson.JSONObject;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;
import tech.yunqian.slim.framework.aspectj.annotation.ValidateRequest;
import tech.yunqian.slim.framework.web.AjaxResult;
import tech.yunqian.slim.framework.web.Response;
import tech.yunqian.slim.project.domain.Device;
import tech.yunqian.slim.project.service.DeviceService;

@Slf4j
@RestController
@RequestMapping("/api/kafka")
public class KafkaController {
    @Value("${topic.telemetry:telemetry-to-gatekeeper}")
    private String telemetryTopic;

    @Value("${topic.event:event-to-gatekeeper}")
    private String eventTopic;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private DeviceService deviceService;

    @ValidateRequest
    @PostMapping(value = "/send", produces = "application/json;charset=UTF-8")
    public AjaxResult Send(@RequestBody JSONObject request) {
        String topic = request.getString("topic");
        String message = request.getString("message");
        kafkaTemplate.send(topic, message);
        return Response.OK.toAjaxResult();
    }

    private Map<String, JSONObject> parseMessage(ConsumerRecord<?, ?> record) {
        Map<String, JSONObject> parsed = new HashMap<>();
        Headers headers = record.headers();
        Iterator<Header> iterator = headers.iterator();
        JSONObject metadata = new JSONObject();
        while (iterator.hasNext()) {
            Header header = iterator.next();
            String key = header.key();
            try {
                String value = new String(header.value(), "utf-8");
                metadata.put(key, value);
            } catch (Exception e) {
                log.info(e.getMessage());
                StackTraceElement[] traces = e.getStackTrace();
                for (StackTraceElement trace : traces) {
                    log.debug(trace.toString());
                }
            }
        }
        JSONObject data = JSONObject.parseObject(record.value().toString());
        parsed.put("data", data);
        parsed.put("metadata", metadata);
        return parsed;
    }

    private Device lookupDevice(String tbDeviceName) {
        Device device = null;
        try {
            device = deviceService.lookupByTerminalTbDeviceName(tbDeviceName);
        } catch (Exception e) {
            log.info(e.getMessage());
            StackTraceElement[] traces = e.getStackTrace();
            for (StackTraceElement trace : traces) {
                log.debug(trace.toString());
            }
        }
        return device;
    }

    private void handleTelemetryTopic(ConsumerRecord<?, ?> record) {
        Map<String, JSONObject> parsed = parseMessage(record);
        JSONObject metadata = parsed.get("metadata");
        JSONObject data = parsed.get("data");
        log.debug("{}[{}]: {} {}", record.topic(), record.partition(), metadata, data);
        
        String tbDeviceName = metadata.getString("tb_msg_md_deviceName");
        if (tbDeviceName == null) {
            return;
        }
        Device device = lookupDevice(tbDeviceName);
        log.debug("device: {}", device);
        if (device == null) {
            return;
        }
        try {
            JSONObject timeSeries = deviceService.getLatestTimeSeries(device);
            Boolean active = data.getBoolean("__active__");
            if (active != null) {
                if (timeSeries == null) {
                    timeSeries = new JSONObject();
                }
                timeSeries.put("__active__" + device.getActiveNumber(), active);
            }
            if (timeSeries == null) {
                return;
            }
            JSONObject cooked = deviceService.cookTelemetries(device, timeSeries);
            log.debug("cooked: {}", cooked);
            deviceService.deliverTelemetries(device, cooked);
            // TODO: [zy] crack from here
            Integer projectId = device.getProjectId();
            if (projectId != null && projectId != 0) {
                deviceService.probeTelemetryTriggers(device, cooked);
            }
        } catch (Exception e) {
            log.info(e.getMessage());
            StackTraceElement[] traces = e.getStackTrace();
            for (StackTraceElement trace : traces) {
                log.debug(trace.toString());
            }
        }
    }

    private void handleEventTopic(ConsumerRecord<?, ?> record) {
        Map<String, JSONObject> parsed = parseMessage(record);
        JSONObject metadata = parsed.get("metadata");
        JSONObject data = parsed.get("data");
        log.debug("{}[{}]: {} {}", record.topic(), record.partition(), metadata, data);
        String tbDeviceName = metadata.getString("tb_msg_md_deviceName");
        if (tbDeviceName == null) {
            return;
        }
        Device device = lookupDevice(tbDeviceName);
        log.debug("device: {}", device);
        if (device == null) {
            return;
        }
        try {
            deviceService.deliverEvents(device, data);
            Integer projectId = device.getProjectId();
            if (projectId != null && projectId != 0) {
                deviceService.probeEventTriggers(device, data);
            }
        } catch (Exception e) {
            log.info(e.getMessage());
            StackTraceElement[] traces = e.getStackTrace();
            for (StackTraceElement trace : traces) {
                log.debug(trace.toString());
            }
        }
    }

    @KafkaListener(topics = "#{'${topic.telemetry:telemetry-to-gatekeeper},${topic.event:event-to-gatekeeper}'.split(',')}")
    public void Receive(ConsumerRecord<?, ?> record) {
        log.info("{}", record);
        String topic = record.topic();
        if (topic.equals(telemetryTopic)) {
            handleTelemetryTopic(record);
        } else if (topic.equals(eventTopic)) {
            handleEventTopic(record);
        }
    }
}